package cn.tedu.charging.order.rabbitmq;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@Slf4j
public class Consumer {
    //消费监听消息
    @Test
    public void consumeQueue01() throws IOException, InterruptedException {
        //1.拿到channel
        Channel channel = ChannelManager.createChannel();
        //2.调用消费方法 持续运行监听queue01中的消息 进行消费
        //String queue: 监听队列名称
        //boolean autoAck: 消费接收 立刻确认(告诉队列删除消息,我已经消费完了)
        //DeliverCallback deliver: 消费回调,处理消费逻辑
        //CancelCallback cancel: 消费取消的回调,当消费者停止 拒绝消息的时候 调用
        DeliverCallback deliver=new DeliverCallback() {
            /**
             * 当rabbitmq 给消费者客户端连接推送过来一条信息message的时候
             * handle被调用一次
             * handle方法 主要作用就是处理消费消息 底层消费入口 类似messageArrived作用
             * String consumerTag: 当前建立连接 监听队列的消费者身份
             * Delivery message: 消息对象,发送的内容 体 头 路由 包含在这个数据中,可以用来做核心消费计算
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                log.info("消费者开始处理消费入口");
                //1.消息体
                byte[] body=message.getBody();
                //2.消息头
                AMQP.BasicProperties properties = message.getProperties();
                //2.1固定头 appId type
                String appId = properties.getAppId();
                String type = properties.getType();
                log.info("appId:{}",appId);
                log.info("type:{}",type);
                //2.2自定义头 headers
                Map<String,Object> headers = properties.getHeaders();
                for (Map.Entry<String, Object> entry : headers.entrySet()) {
                    log.info("key:{},value:{}",entry.getKey(),entry.getValue());
                }
                //3.消息路由
                String routingKey = message.getEnvelope().getRoutingKey();
                log.info("routingKey:{}",routingKey);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            /*
                当这个消费者连接 取消监听订阅队列的时候 cancel 会调用handle
                一般是因为消费者为了能够处理消费消息,会在项目进程启动时,初始化一些资源,比如各种连接
                一旦取消监听,这些资源就应该释放掉,使用handle方法释放
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                log.info("消费者取消监听队列queue01,释放资源");
            }
        };
        channel.basicConsume(
                "queue01",
                true,
                deliver,
                cancel
        );
        //保持线程不死,持续运行
        Thread.sleep(1000000);
    }

    //消费自动确认案例 消费监听queue1
    @Test
    public void consumeQueue1AutoAck() throws Exception {
        Channel channel = ChannelManager.createChannel();
        channel.basicConsume(
                "queue1",
                true,
                (consumerTag,message)->{
                    log.info("自动确认获取消息");
                },
                consumerTag->{});
        Thread.sleep(1000000);
    }
    //手动确认 关闭自动确认,在执行完消息的时候 反馈确认结果
    @Test
    public void consumeQueue1ManualAck() throws Exception {
        Channel channel = ChannelManager.createChannel();
        channel.basicConsume(
                "queue1",
                false,
                (consumerTag,message)->{
                    log.info("手动确认获取消息");
                    //什么都不做,每条消息 卡在unacked
                    //basicAck调用 就可以告诉rabbitmq这条消息具体消费ack成功了
                    channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                },
                consumerTag->{});
        Thread.sleep(1000000);
    }
    //产生死信的消费逻辑
    @Test
    public void consumeAndNack() throws Exception {
        Channel channel = ChannelManager.createChannel();
        channel.basicConsume(
                "biz_q",
                false,
                (consumerTag,message)->{
                    log.info("手动确认获取消息,但是消息拒绝且不允许重新入队");
                    //反馈拒绝
                    channel.basicNack(
                            message.getEnvelope().getDeliveryTag(),
                            false,
                            false);
                },
                consumerTag->{});
        Thread.sleep(1000000);
    }

}
